package com.yc.bigdata.flink.demo;

import com.yc.bigdata.flink.demo.data.WebLogData;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.util.Collector;

/**
 * <p></p>
 *
 * @author: YuanChilde
 * @date: 2020-02-10 16:18
 * @version: 1.0
 * Modification History:
 * Date    Author      Version     Description
 * -----------------------------------------------------------------
 * 2020-02-10 16:18    YuanChilde     1.0        新增
 */
public class DataSetWebLogAnalysis {

    public static void main(String[] args) throws Exception {
        final ParameterTool params = ParameterTool.fromArgs(args);
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(params);

        DataSet<Tuple2<String, String>> documents = getDocumentsDataSet(env, params);
        DataSet<Tuple3<Integer, String, Integer>> ranks = getRanksDataSet(env, params);
        DataSet<Tuple2<String, String>> visits = getVisitsDataSet(env, params);

        // 按关键字过滤数据
        DataSet<Tuple1<String>> filterDocs = documents
                .filter(new FilterDocByKeyWords())
                .project(0);

        // 按日期过滤数据
        DataSet<Tuple1<String>> filterVisits = visits
                .filter(new FilterVisitsByDate())
                .project(0);

        // 按分数过滤数据
        DataSet<Tuple3<Integer, String, Integer>> filterRanks = ranks
                .filter(new FilterByRank());

        DataSet<Tuple3<Integer, String, Integer>> joinDocsRanks =
                filterDocs.join(filterRanks)
                        .where(0).equalTo(1)
                        .projectSecond(0, 1, 2);

        // Anti-join urls with visits, i.e., retain all URLs which have NOT been visited in a certain time
        DataSet<Tuple3<Integer, String, Integer>> result =
                joinDocsRanks.coGroup(filterVisits)
                        .where(1).equalTo(0)
                        .with(new AntiJoinVisits());

        if (params.has("output")) {
            result.writeAsCsv(params.get("output"), "\n", "|");
            // execute program
            env.execute("WebLogAnalysis Example");
        } else {
            System.out.println("Printing result to stdout. Use --output to specify output path.");
            result.print();
        }
    }

    public static class FilterDocByKeyWords implements FilterFunction<Tuple2<String, String>> {

        private static final String[] KEYWORDS = { " editors ", " oscillations " };

        /**
         * Filters for documents that contain all of the given keywords and projects the records on the URL field.
         *
         * <p>Output Format:
         * 0: URL
         * 1: DOCUMENT_TEXT
         */
        @Override
        public boolean filter(Tuple2<String, String> value) throws Exception {
            // FILTER
            // Only collect the document if all keywords are contained
            String docText = value.f1;
            for (String kw : KEYWORDS) {
                if (!docText.contains(kw)) {
                    return false;
                }
            }
            return true;
        }
    }

    public static class FilterByRank implements FilterFunction<Tuple3<Integer, String, Integer>> {

        private static final int RANKFILTER = 40;

        /**
         * Filters for records of the rank relation where the rank is greater
         * than the given threshold.
         *
         * <p>Output Format:
         * 0: RANK
         * 1: URL
         * 2: AVG_DURATION
         */
        @Override
        public boolean filter(Tuple3<Integer, String, Integer> value) throws Exception {
            return (value.f0 > RANKFILTER);
        }
    }

    /**
     * MapFunction that filters for records of the visits relation where the year
     * (from the date string) is equal to a certain value.
     */
    public static class FilterVisitsByDate implements FilterFunction<Tuple2<String, String>> {

        private static final int YEARFILTER = 2007;

        /**
         * Filters for records of the visits relation where the year of visit is equal to a
         * specified value. The URL of all visit records passing the filter is emitted.
         *
         * <p>Output Format:
         * 0: URL
         * 1: DATE
         */
        @Override
        public boolean filter(Tuple2<String, String> value) throws Exception {
            // Parse date string with the format YYYY-MM-DD and extract the year
            String dateString = value.f1;
            int year = Integer.parseInt(dateString.substring(0, 4));
            return (year == YEARFILTER);
        }
    }


    /**
     * CoGroupFunction that realizes an anti-join.
     * If the first input does not provide any pairs, all pairs of the second input are emitted.
     * Otherwise, no pair is emitted.
     */
    public static class AntiJoinVisits implements CoGroupFunction<Tuple3<Integer, String, Integer>, Tuple1<String>, Tuple3<Integer, String, Integer>> {

        /**
         * If the visit iterator is empty, all pairs of the rank iterator are emitted.
         * Otherwise, no pair is emitted.
         *
         * <p>Output Format:
         * 0: RANK
         * 1: URL
         * 2: AVG_DURATION
         */
        @Override
        public void coGroup(Iterable<Tuple3<Integer, String, Integer>> ranks, Iterable<Tuple1<String>> visits, Collector<Tuple3<Integer, String, Integer>> out) {
            if (!visits.iterator().hasNext()) {
                for (Tuple3<Integer, String, Integer> next : ranks) {
                    out.collect(next);
                }
            }
        }
    }

    private static DataSet<Tuple2<String, String>> getDocumentsDataSet(ExecutionEnvironment env, ParameterTool params) {
        // Create DataSet for documents relation (URL, Doc-Text)
        if (params.has("documents")) {
            return env.readCsvFile(params.get("documents"))
                    .fieldDelimiter("|")
                    .types(String.class, String.class);
        } else {
            System.out.println("Executing WebLogAnalysis example with default documents data set.");
            System.out.println("Use --documents to specify file input.");
            return WebLogData.getDocumentDataSet(env);
        }
    }

    private static DataSet<Tuple3<Integer, String, Integer>> getRanksDataSet(ExecutionEnvironment env, ParameterTool params) {
        // Create DataSet for ranks relation (Rank, URL, Avg-Visit-Duration)
        if (params.has("ranks")) {
            return env.readCsvFile(params.get("ranks"))
                    .fieldDelimiter("|")
                    .types(Integer.class, String.class, Integer.class);
        } else {
            System.out.println("Executing WebLogAnalysis example with default ranks data set.");
            System.out.println("Use --ranks to specify file input.");
            return WebLogData.getRankDataSet(env);
        }
    }

    private static DataSet<Tuple2<String, String>> getVisitsDataSet(ExecutionEnvironment env, ParameterTool params) {
        // Create DataSet for visits relation (URL, Date)
        if (params.has("visits")) {
            return env.readCsvFile(params.get("visits"))
                    .fieldDelimiter("|")
                    .includeFields("011000000")
                    .types(String.class, String.class);
        } else {
            System.out.println("Executing WebLogAnalysis example with default visits data set.");
            System.out.println("Use --visits to specify file input.");
            return WebLogData.getVisitDataSet(env);
        }
    }
}
